package com.jeffse.light.stream.core.connector.pulsar;

import com.jeffse.light.stream.core.data.context.SourceContext;
import com.jeffse.light.stream.core.function.base.AbstractSourceFunction;
import com.jeffse.light.stream.core.utils.PulsarUtils;
import org.apache.pulsar.client.api.PulsarClientException;

import java.util.ArrayList;
import java.util.List;

public class PulsarSource<T extends PulsarListener> implements AbstractSourceFunction<T> {
    private String serviceUrl = "";
    private String topic = "";
    private List<String> topics = new ArrayList<>();
    private String group = "";
    private Class<T> pulsarListenerClass = null;

    public PulsarSource(String serviceUrl, String topic, String group, Class<T> pulsarListenerClass) throws PulsarClientException {
        this.serviceUrl = serviceUrl;
        this.topic = topic;
        this.group = group;
        this.pulsarListenerClass = pulsarListenerClass;
    }

    public PulsarSource(String serviceUrl, List<String> topics, String group, Class<T> pulsarListenerClass) throws PulsarClientException {
        this.serviceUrl = serviceUrl;
        this.topics = topics;
        this.group = group;
        this.pulsarListenerClass = pulsarListenerClass;
    }

    @Override
    public void run(SourceContext<T> ctx) throws Exception {
        PulsarUtils.getINSTANCE().getClient(serviceUrl);
        PulsarListener pulsarListener = pulsarListenerClass.newInstance();
        pulsarListener.setCtx(ctx);
        if(topic.equals("")){
            PulsarUtils.getINSTANCE().getConsumer(topics, group, pulsarListener);
        }else{
            PulsarUtils.getINSTANCE().getConsumer(topic, group, pulsarListener);
        }

    }

    @Override
    public void cancel() {
        PulsarUtils.getINSTANCE().removeConsumer();
    }
}
